package com.ygl.hbase;

import com.ygl.hbase.pojo.*;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Date 2022/1/24 14:28
 * @Created yanggld
 * @Description
 */
public class HbaseUtil {

    private Connection connection;
    private Admin admin;

    public HbaseUtil(String zkQuorum) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum);
        Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();
        this.connection = connection;
        this.admin = admin;
    }

    public HbaseUtil(Map<String, String> params) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        for (Map.Entry<String, String> entry : params.entrySet()) {
            conf.set(entry.getKey(), entry.getValue());
        }
        Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();
        this.admin = admin;
    }

    /**
     * 创建命名空间
     *
     * @param namespace
     */
    public void createNamespace(String namespace) throws Exception {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
        try {
            admin.createNamespace(namespaceDescriptor);
        } catch (NamespaceExistException e) {
            throw new RuntimeException("namespace：[" + namespace + "] is exists");
        }
    }

    /**
     * 表是是否存在
     *
     * @param namespace 命名空间名
     * @param table     表名
     * @return
     * @throws IOException
     */
    public boolean tableExists(String namespace, String table) throws IOException {
        TableName tableName = TableName.valueOf(namespace, table);
        return admin.tableExists(tableName);
    }

    /**
     * 表是是否存在
     *
     * @param table 表名
     * @return
     * @throws IOException
     */
    public boolean tableExists(String table) throws IOException {
        TableName tableName = TableName.valueOf(table);
        return admin.tableExists(tableName);
    }

    /**
     * 创建表
     *
     * @param table 表名
     * @param cfs   列族
     */
    public void createTable(String table, String... cfs) throws IOException {
        if (cfs.length == 0) {
            throw new RuntimeException("cfs is emtpy");
        }
        if (tableExists(table)) {
            throw new RuntimeException("table: " + table + " exists");
        }
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(table));
        for (String cf : cfs) {
            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(cf.getBytes()).build();
            tableDescriptorBuilder.setColumnFamily(familyDescriptor);
        }
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
        admin.createTable(tableDescriptor);
    }

    /**
     * 显示表列表
     * @return
     * @throws IOException
     */
    public TableName[] listTableNames() throws IOException {
        return admin.listTableNames();
    }

    /**
     * 显示表列表
     * @param nsName 命名空间
     * @return
     * @throws IOException
     */
    public TableName[] listTableNames(String nsName) throws IOException {
        return admin.listTableNamesByNamespace(nsName);
    }

    /**
     * 删除表
     *
     * @param table 表名
     * @throws IOException
     */
    public void dropTable(String table) throws IOException {
        if (!tableExists(table)) {
            throw new RuntimeException("table: " + table + " is not exists");
        }
        TableName tableName = TableName.valueOf(table);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }

    /**
     * 清空表数据
     *
     * @param table 表名
     * @param preserveSplits
     * @throws IOException
     */
    public void truncateTable(String table, boolean preserveSplits) throws IOException {
        if (!tableExists(table)) {
            throw new RuntimeException("table: " + table + " is not exists");
        }
        TableName tableName = TableName.valueOf(table);
        admin.disableTable(tableName);
        admin.truncateTable(tableName, preserveSplits);
    }

    /**
     * 关闭资源
     *
     * @throws IOException
     */
    public void close() throws IOException {
        if (admin != null) {
            admin.close();
        }
        if (connection != null) {
            connection.close();
        }
    }

    /**
     * 保存数据
     *
     * @param tableName 表名
     * @param rowKey
     * @param cf        列簇
     * @param cn        列名
     * @param value     数据
     */
    public void putData(String tableName, String rowKey, String cf, String cn, String value) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));
        table.put(put);
    }

    /**
     * 通过rowKey查询数据
     *
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public HBaseModel getData(String tableName, String rowKey) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        Cell[] cells = result.rawCells();
        Map<String, List<Cell>> cellsGrouped = groupByCf(cells);
        List<Map<String, String>> rows = new ArrayList<>(cells.length);
        for (Map.Entry<String, List<Cell>> entry : cellsGrouped.entrySet()) {
            String cf = entry.getKey();
            List<Cell> cellList = entry.getValue();
            Map<String, String> data = new HashMap<>();
            data.put(HBaseModel.ROW_KEY, rowKey);
            data.put(HBaseModel.CF, cf);
            for (Cell cell : cellList) {
                String cn = Bytes.toString(CellUtil.cloneQualifier(cell));
                String val = Bytes.toString(CellUtil.cloneValue(cell));
                data.put(cn, val);
            }
            rows.add(data);
        }
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        model.setRows(rows);
        return model;
    }


    private List<Map<String, String>> cellsGrouped2Rows(Map<String, List<Cell>> cellsGrouped) {
        List<Map<String, String>> rows = new ArrayList<>();
        for (Map.Entry<String, List<Cell>> entry : cellsGrouped.entrySet()) {
            String cf = entry.getKey();
            List<Cell> cellList = entry.getValue();
            Map<String, String> data = cells2Row(cf, cellList);
            rows.add(data);
        }
        return rows;
    }

    private Map<String, String> cells2Row(String cf, List<Cell> cells) {
        Map<String, String> data = new HashMap<>();
        for (Cell cell : cells) {
            data.put(HBaseModel.CF, cf);
            String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
            if (data.get(HBaseModel.ROW_KEY) == null) {
                data.put(HBaseModel.ROW_KEY, rowKey);
            }
            String cn = Bytes.toString(CellUtil.cloneQualifier(cell));
            String val = Bytes.toString(CellUtil.cloneValue(cell));
            data.put(cn, val);
        }
        return data;
    }

    /**
     * 通过rowKey,cf查询数据
     *
     * @param tableName
     * @param rowKey
     * @param cf        列簇
     * @return
     * @throws IOException
     */
    public HBaseModel getData(String tableName, String rowKey, String cf) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addFamily(Bytes.toBytes(cf));
        Result result = table.get(get);
        List<Cell> cellList = result.listCells();
        Map<String, String> row = cells2Row(cf, cellList);
        List<Map<String, String>> rows = new ArrayList<>();
        rows.add(row);
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        model.setRows(rows);
        return model;
    }

    /**
     * 通过rowKey,cf,cn查询数据
     *
     * @param tableName
     * @param rowKey
     * @param cf        列簇
     * @param cn        列名
     * @return
     * @throws IOException
     */
    public HBaseModel getData(String tableName, String rowKey, String cf, String cn) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn));
        Result result = table.get(get);
        List<Cell> cellList = result.listCells();
        Map<String, String> row = cells2Row(cf, cellList);
        List<Map<String, String>> rows = new ArrayList<>();
        rows.add(row);
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        model.setRows(rows);
        return model;
    }

    /**
     * 查询全表数据
     *
     * @param tableName
     * @return
     * @throws IOException
     */
    public HBaseModel getDataList(String tableName) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        List<Map<String, String>> rows = new ArrayList<>();
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            Map<String, List<Cell>> cellsGroupedByCf = groupByCf(cells);
            List<Map<String, String>> rowList = cellsGrouped2Rows(cellsGroupedByCf);
            rows.addAll(rowList);
        }
        model.setRows(rows);
        return model;
    }

    /**
     * 查询数据
     *
     * @param tableName
     * @param startRowKey    开始rowkey
     * @param inclusiveStart 是否包含开始key
     * @param stopRowKey     结束rowkey
     * @param inclusiveStop  是否包含结束key
     * @return
     * @throws IOException
     */
    public HBaseModel getDataList(String tableName, String startRowKey, boolean inclusiveStart, String stopRowKey, boolean inclusiveStop) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        if (StringUtils.isNotEmpty(startRowKey)) {
            scan.withStartRow(Bytes.toBytes(startRowKey), inclusiveStart);
        }
        if (StringUtils.isNotEmpty(stopRowKey)) {
            scan.withStopRow(Bytes.toBytes(stopRowKey), inclusiveStop);
        }
        ResultScanner scanner = table.getScanner(scan);
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        List<Map<String, String>> rows = new ArrayList<>();
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            Map<String, List<Cell>> cellsGroupedByCf = groupByCf(cells);
            List<Map<String, String>> rowList = cellsGrouped2Rows(cellsGroupedByCf);
            rows.addAll(rowList);
        }
        model.setRows(rows);
        return model;
    }

    /**
     * 通过id或者ids 查询单表数据
     * 可扩展带列簇和带列查询
     *
     * @param getModel
     * @return
     * @throws IOException
     */
    public HBaseModel getDataList(HBaseGetModel getModel) throws IOException {
        String tableName = getModel.getTableName();
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<String> rowkeys = getModel.getRowKeys();
        List<Get> getList = new ArrayList<>();
        for (String rowKey : rowkeys) {
            Get get = new Get(Bytes.toBytes(rowKey));
            getList.add(get);
        }
        Result[] results = table.get(getList);
        return buildModel(tableName, results);
    }

    /**
     * 通过id或者ids 查询多表数据
     * 可扩展带列簇和带列查询
     *
     * @param getModels
     * @return
     * @throws IOException
     */
    public List<HBaseModel> getDataList(List<HBaseGetModel> getModels) throws IOException {
        List<HBaseModel> list = new ArrayList<>(getModels.size());
        for (HBaseGetModel getModel : getModels) {
            HBaseModel hBaseModel = getDataList(getModel);
            list.add(hBaseModel);
        }
        return list;
    }

    /**
     * 通过key删除所有数据
     *
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public void deleteData(String tableName, String rowKey) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
    }

    /**
     * 通过key删除列簇数据
     *
     * @param tableName
     * @param rowKey
     * @param cf        列簇
     * @throws IOException
     */
    public void deleteData(String tableName, String rowKey, String cf) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addFamily(Bytes.toBytes(cf));
        table.delete(delete);
    }

    /**
     * 通过key删除列数据
     *
     * @param tableName
     * @param rowKey
     * @param cf
     * @param cn
     * @throws IOException
     */
    public void deleteData(String tableName, String rowKey, String cf, String cn) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(cn));
        table.delete(delete);
    }

    /**
     * 通过key或者keys 删除单表所有数据
     *
     * @param getModel
     * @throws IOException
     */
    public void deleteData(HBaseGetModel getModel) throws IOException {
        String tableName = getModel.getTableName();
        List<String> rowKeys = getModel.getRowKeys();
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<Delete> deleteList = new ArrayList<>(rowKeys.size());
        for (String rowKey : rowKeys) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            deleteList.add(delete);
        }
        table.delete(deleteList);
    }

    /**
     * 通过key或者keys 删除多表所有数据
     *
     * @param getModels
     * @throws IOException
     */
    public void deleteData(List<HBaseGetModel> getModels) throws IOException {
        for (HBaseGetModel getModel : getModels) {
            deleteData(getModel);
        }
    }


    /**
     * 保存单表数据
     *
     * @param tableModel
     * @throws IOException
     */
    public void putData(HBaseModel tableModel) throws IOException {
        String tableName = tableModel.getTableName();
        if (StringUtils.isEmpty(tableName)) {
            throw new RuntimeException("tableName is empty!");
        }
        List<Map<String, String>> rows = tableModel.getRows();
        if (CollectionUtils.isEmpty(rows)) {
            throw new RuntimeException("rows is empty!");
        }
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<Put> putList = new ArrayList<>(rows.size());
        Set<String> cfs = new HashSet<>();
        for (Map<String, String> row : rows) {
            Object rowKeyObj = row.get(HBaseModel.ROW_KEY);
            if (rowKeyObj == null) {
                throw new RuntimeException("rowKey is null, map is :" + row);
            }
            Object cfObj = row.get(HBaseModel.CF);
            if (cfObj == null) {
                throw new RuntimeException("cf is null, map is :" + row);
            }
            String rowKey = rowKeyObj.toString();
            String cf = cfObj.toString();
            cfs.add(cf);
            Put put = new Put(Bytes.toBytes(rowKey));
            for (Map.Entry<String, String> entry : row.entrySet()) {
                String key = entry.getKey();
                if (!HBaseModel.ROW_KEY.equals(key) && !HBaseModel.CF.equals(key)) {
                    Object value = entry.getValue();
                    put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(key), getBytes(value));
                }
            }
            putList.add(put);
        }
        if (!tableExists(tableName)) {
            createTable(tableName, cfs.toArray(new String[cfs.size()]));
        }
        table.put(putList);
    }

    /**
     * 保存多表数据
     *
     * @param tableModels
     * @throws IOException
     */
    public void putData(List<HBaseModel> tableModels) throws IOException {
        for (HBaseModel tableModel : tableModels) {
            putData(tableModel);
        }
    }

    /**
     * BinaryComparator          // 匹配完整字节数组
     * BinaryPrefixComparator    // 匹配字节数组前缀
     * BitComparator // 按位执行与、或、异或比较
     * NullComparator // 判断当前值是不是 NULL
     * RegexStringComparator     // 正则表达式匹配字符串
     * SubstringComparator       // 子串匹配，相当于 contains()
     *
     *
     * 行键过滤器 RowFilter
     * 列族过滤器 FamilyFilter
     * 列名过滤器 QualifierFilter
     * 值过滤器 ValueFilter
     * 参考列过滤器 DependentColumnFilter
     * 单列值过滤器 SingleColumnValueFilter
     * 单列值排除器 SingleColumnValueExcludeFilter
     * 行前缀过滤器 PrefixFilter
     * 列前缀过滤器 ColumnPrefixFilter
     * 分页过滤器 PageFilter
     * 行键过滤器 KeyOnlyFilter
     * 首次行键过滤器 FirstKeyOnlyFilter
     * 包含结束的过滤器 InclusiveStopFilter
     * 时间戳过滤器 TimestampsFilter
     * 列计数过滤器 ColumnCountGetFilter
     * 列分页过滤器 ColumnPaginationFilter
     * 跳转过滤器 SkipFilter
     * 全匹配过滤 WhileMatchFilter
     * 多种过滤条件的使用方法
     */
    /**
     * 通过过滤器查询数据
     * @param tableName
     * @param rowFilter
     * @return
     * @throws IOException
     */
    public HBaseModel filter(String tableName,Filter rowFilter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        Table table = connection.getTable(TableName.valueOf(tableName));
        ResultScanner scanner = table.getScanner(scan);
        return buildModel(tableName, scanner);
    }

    /**
     * Result集合 转 HBaseModel
     * @param tableName
     * @param list
     * @return
     */
    public HBaseModel buildModel(String tableName,Iterable<Result> list){
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        List<Map<String, String>> rows = new ArrayList<>();
        for (Result result : list) {
            Cell[] cells = result.rawCells();
            Map<String, List<Cell>> cellsGroupedByCf = groupByCf(cells);
            List<Map<String, String>> rowList = cellsGrouped2Rows(cellsGroupedByCf);
            rows.addAll(rowList);
        }
        model.setRows(rows);
        return model;
    }

    /**
     * Result数组 转 HBaseModel
     * @param tableName
     * @param array
     * @return
     */
    public HBaseModel buildModel(String tableName,Result[] array){
        HBaseModel model = new HBaseModel();
        model.setTableName(tableName);
        List<Map<String, String>> rows = new ArrayList<>();
        for (Result result : array) {
            Cell[] cells = result.rawCells();
            Map<String, List<Cell>> cellsGroupedByCf = groupByCf(cells);
            List<Map<String, String>> rowList = cellsGrouped2Rows(cellsGroupedByCf);
            rows.addAll(rowList);
        }
        model.setRows(rows);
        return model;
    }

    /**
     * cells 通过列簇分组
     * @param cells
     * @return
     */
    private Map<String, List<Cell>> groupByCf(Cell[] cells) {
        return Arrays.stream(cells).collect(Collectors.groupingBy(cell -> Bytes.toString(CellUtil.cloneFamily(cell))));
    }


    public byte[] getBytes(Object obj) {
        if (obj == null) {
            return null;
        }
        if (obj instanceof Integer) {
            return Bytes.toBytes((int) obj);
        } else if (obj instanceof Float) {
            return Bytes.toBytes((float) obj);
        } else if (obj instanceof Long) {
            return Bytes.toBytes((long) obj);
        } else if (obj instanceof Double) {
            return Bytes.toBytes((double) obj);
        } else if (obj instanceof String) {
            return Bytes.toBytes((String) obj);
        } else if (obj instanceof Short) {
            return Bytes.toBytes((short) obj);
        } else if (obj instanceof Boolean) {
            return Bytes.toBytes((boolean) obj);
        } else if (obj instanceof BigDecimal) {
            return Bytes.toBytes((BigDecimal) obj);
        }
        return null;
    }

}
